package kp.rpc.server.client;

import cn.hutool.core.util.IdUtil;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetClient;
import io.vertx.core.net.NetSocket;
import kp.rpc.Enum.ProtocolMessageSerializerEnum;
import kp.rpc.Enum.ProtocolMessageTypeEnum;
import kp.rpc.RpcApplication;
import kp.rpc.model.RpcRequest;
import kp.rpc.model.RpcResponse;
import kp.rpc.model.ServiceMetaInfo;
import kp.rpc.protocol.ProtocolConstant;
import kp.rpc.protocol.ProtocolMessage;
import kp.rpc.protocol.encode.ProtocolMessageDecoder;
import kp.rpc.protocol.encode.ProtocolMessageEncoder;
import kp.rpc.server.wrapper.TcpBufferHandlerWrapper;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

@Slf4j
public class VertxTcpClient {
    public static RpcResponse doRpcResponse(RpcRequest rpcRequest, ServiceMetaInfo serviceMetaInfo) throws ExecutionException, InterruptedException {
        // 发送tcp请求
        Vertx vertx = Vertx.vertx();
        //创建tcp请求客户端
        NetClient netClient = vertx.createNetClient();
        CompletableFuture<RpcResponse> future = new CompletableFuture<>();
        netClient.connect(serviceMetaInfo.getServicePort(), serviceMetaInfo.getServiceHost(), result -> {
            if (result.succeeded()) {
                log.info("connect to TCP server");
                //拿到socket管道
                NetSocket socket = result.result();
                //发送数据
                //构造信息
                ProtocolMessage<RpcRequest> requestProtocolMessage = new ProtocolMessage<>();
                ProtocolMessage.Header header = new ProtocolMessage.Header();
                header.setMagic(ProtocolConstant.PROTOCOL_MAGIC);
                header.setVersion(ProtocolConstant.PROTOCOL_VERSION);
                //设置序列化器
                header.setSerializer((byte) Objects.requireNonNull(ProtocolMessageSerializerEnum.getEnumByValue(RpcApplication.getRpcConfig().getSerializer())).getKey());
                //设置请求类型
                header.setType((byte) ProtocolMessageTypeEnum.REQUEST.getKey());
                //设置雪花id
                header.setRequestId(IdUtil.getSnowflakeNextId());
                requestProtocolMessage.setHeader(header);
                requestProtocolMessage.setBody(rpcRequest);
                //编码请求
                try {
                    Buffer requestEncode = ProtocolMessageEncoder.encode(requestProtocolMessage);
                    //写入数据发送请求
                    socket.write(requestEncode);
                } catch (IOException e) {
                    throw new RuntimeException("协议消息编码错误!");
                }
                TcpBufferHandlerWrapper tcpBufferHandlerWrapper = new TcpBufferHandlerWrapper(buffer -> {
                    try {
                        ProtocolMessage<RpcResponse> responseProtocolMessage = (ProtocolMessage<RpcResponse>) ProtocolMessageDecoder.decode(buffer);
                        future.complete(responseProtocolMessage.getBody());
                    } catch (IOException e) {
                        throw new RuntimeException("协议消息解码错误!");
                    }
                });
                //接收响应,监听socket管道
                socket.handler(tcpBufferHandlerWrapper);
            } else {
                log.info("Failed to connect to TCP server");
            }
        });
        RpcResponse rpcResponse = future.get();
        //关闭tcp客户端
        netClient.close();
        //get,阻塞
        return rpcResponse;
    }
}
